package br.edu.ufcg.greengrid.workload;

import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import com.mallardsoft.tuple.Pair;
import com.mallardsoft.tuple.Triple;
import com.mallardsoft.tuple.Tuple;

public class DGTAConverter implements PlainTextConverter {

	private static final Long NOT_DELIVERED = -1L;

	private Map<String, Pair<Long, Long>> data; // <MachineId, Pair<Timestamp, Idletime>>

	private Map<String, Long> lastTimestamp; // <MachineId, Timestamp>

	private BulkData<Triple<String, Long, Long>> bulkData;

	public DGTAConverter(BulkData<Triple<String, Long, Long>> bulkData) {

		super();
		this.bulkData = bulkData;
		this.data = new LinkedHashMap<String, Pair<Long, Long>>();
		this.lastTimestamp = new HashMap<String, Long>();
	}

	@Override
	public void convert(String fileName, String line) {

		String[] splitFileName = fileName.split("\\.");

		StringBuffer sbMachine = new StringBuffer();
		for (int i = 0; i < splitFileName.length - 1; i++) {
			sbMachine.append(splitFileName[i]);
		}
		String machine = sbMachine.toString();

		convertForMachine(machine, line);

	}

	private void convertForMachine(String cMachine, String line) {

		String[] splitLine = line.split(" ");
		int index = 0;
		double epochStartTimeD = Double.parseDouble(splitLine[index++]);
		double epochStopTimeD = Double.parseDouble(splitLine[index++]);
		double operationDeliveredD = Double.parseDouble(splitLine[index++]);
		
		Long epochStartTime = (long) Math.ceil(epochStartTimeD);
		Long epochStopTime = (long) Math.ceil(epochStopTimeD);
		Long operationDelivered = (long) Math.ceil(operationDeliveredD);

		// The "s" means "stored"
		Pair<Long, Long> pair = this.data.get(cMachine);
		String sMachine = null;
		Long sTimestamp = null;
		Long sIdletime = null;

		if (pair != null) {
			sMachine = cMachine;
			sTimestamp = Tuple.get1(pair);
			sIdletime = Tuple.get2(pair);
		}

		Long delta = 0L;
		Long lastTimestamp = this.lastTimestamp.get(cMachine);
		if (lastTimestamp != null && lastTimestamp.compareTo(epochStartTime) > 0) {
			throw new IllegalArgumentException(String.format(
					"Error! Inserting observations with no increasing timestamp is not allowed!\n" +
							"Last timestamp: %d \n" + 
							"Current epochStartTime(long): %d\n" + 
							"Current epochStopTime(long): %d\n" +
							"Current operationDelivered(long): %d\n" +
							"Current epochStartTime(double): %f\n" + 
							"Current epochStopTime(double): %f\n" +
							"Current operationDelivered(double): %f\n" +
							"Machine: %s", lastTimestamp, epochStartTime,
					epochStopTime, operationDelivered, epochStartTimeD,
					epochStopTimeD, operationDeliveredD, cMachine));
		} else {
			if (lastTimestamp == null) {
				lastTimestamp = epochStartTime;
			}
			delta = epochStartTime - lastTimestamp;
			epochStartTime -= delta;
			epochStopTime -= delta;
			this.lastTimestamp.put(cMachine, epochStopTime);
		}

		if (sMachine == null && !NOT_DELIVERED.equals(operationDelivered)) {
			sMachine = cMachine;
			sTimestamp = epochStartTime;
			sIdletime = epochStopTime - epochStartTime;
			this.data.put(sMachine, Tuple.from(sTimestamp, sIdletime));
		} else if (sMachine != null && !NOT_DELIVERED.equals(operationDelivered)) {
			sIdletime += epochStopTime - epochStartTime;
			this.data.put(sMachine, Tuple.from(sTimestamp, sIdletime));
		} else if (sMachine != null && NOT_DELIVERED.equals(operationDelivered)) {
			add(sMachine, sTimestamp, sIdletime);
			this.data.remove(sMachine);
		}

	}

	private void add(String machine, Long timestamp, Long idletime) {

		this.bulkData.add(Tuple.from(machine, timestamp, idletime));
	}

	@Override
	public void close() {

		for (Entry<String, Pair<Long, Long>> entry : this.data.entrySet()) {
			String sMachine = entry.getKey();
			Pair<Long, Long> pair = entry.getValue();
			Long sTimestamp = Tuple.get1(pair);
			Long sIdletime = Tuple.get2(pair);
			add(sMachine, sTimestamp, sIdletime);
		}
		this.data.clear();
	}

	@Override
	public List<String> getLines() {

		return this.bulkData.getLines();
	}

}
